/*
 * Copyright (c) 2016 moondev.cn
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package cn.moondev.moonlog.example.moonlog.monitor.mq;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import cn.moondev.moonlog.example.moonlog.monitor.handler.MonitorMsgHandler;
import cn.moondev.moonlog.example.moonlog.monitor.spring.SpringContextHolder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;

import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.message.MessageExt;


public class MessageListenerAdapter implements MessageListenerConcurrently {
	
	private static final Logger log = LoggerFactory.getLogger(MessageListenerAdapter.class);
	
	@Autowired
	public SpringContextHolder springContextHolder;

	@Override
	public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
		// 消息归类
		Map<String, List<String>> tagMap = classifyMessages(msgs);
		// 处理消息
		processMessages(tagMap);
		return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
	}
	
	/**
	 * 处理消息<BR>
	 * 规则：tag为处理该消息实现类的spring bean名称<BR>
	 * 直接实例化tag，然后将消息传递给具体的处理类
	 */
	private void processMessages(Map<String, List<String>> tagMap) {
		log.info("开始消费消息" + tagMap);
		ApplicationContext springContextHolder = SpringContextHolder.applicationContext;
        if (springContextHolder == null) {
        	log.info("springContextHolder == null");
            throw new RuntimeException("please init spring applicationContext or conf ApplicationContextUtils");
        }
		for (Entry<String, List<String>> entry: tagMap.entrySet()) {
			String tag = entry.getKey();
			List<String> msgList = entry.getValue();
			MonitorMsgHandler handler = (MonitorMsgHandler)springContextHolder.getBean(tag);
			handler.handle(msgList);
		}
	}

	/**
	 * 将消息按照tag分类，每一个tag有一列消息
	 * @return
	 */
	private Map<String, List<String>> classifyMessages(List<MessageExt> msgs) {
		// key = tag
		// value = List<msg.body>
		Map<String, List<String>> tagMap = new HashMap<String, List<String>>();
		String tag = null;
		List<String> msgList = null;
		for (MessageExt msg : msgs) {
			tag = msg.getTags();
			msgList = tagMap.get(tag);
			if (msgList == null) {
				msgList = new ArrayList<String>();
				msgList.add(new String(msg.getBody()));
				tagMap.put(tag, msgList);
			} else {
				msgList.add(new String(msg.getBody()));
			}
		}
		return tagMap;
	}
}
